package com.iot08.api.service.supplement;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.iot08.api.config.redis.RedisClusterService;
import com.iot08.api.config.redis.RedisFixService;
import com.iot08.api.kafka.KafkaProducer;
import com.iot08.api.service.AccountService;
import com.iot08.api.service.DeviceService;
import com.iot08.api.service.IotCardInfoService;
import com.iot08.api.utils.buckettool.BaseBucketTool;
import com.iot08.common.constant.CacheEnum;
import com.iot08.common.constant.Const;
import com.iot08.common.constant.ConstCacheField;
import com.iot08.common.constant.ConstKafka;
import com.iot08.common.entity.app.Account;
import com.iot08.common.entity.app.ChatGroup;
import com.iot08.common.entity.app.Device;
import com.iot08.common.entity.app.IotCardInfo;
import com.iot08.common.utils.CollectionUtil;
import com.iot08.common.utils.DateUtil;
import com.iot08.common.utils.RandomUtil;
import com.iot08.common.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

/**
 * @version 1.0
 * @Description: kafka推送相关业务逻辑
 * @time: 2020/12/16 18:15
 * @author: Rock
 */
@Slf4j
@Service
public class KafkaService {

    @Resource
    private KafkaProducer kafkaProducer;
    @Resource
    private RedisClusterService redisClusterService;
    @Resource
    private RedisFixService redisFixService;
    @Resource
    private DeviceService deviceService;

    @Resource
    private AccountService accountService;

    @Resource
    private IotCardInfoService iotCardInfoService;

    @Value("${domain.name}")
    private String domainName;

    @Resource
    private BaseBucketTool baseBucketTool;

    /**
     * 推送
     * @param pushType 推格式 1:群 2:多用户 3:单用户 4:推主题
     * @param id 群ID / 多用户列表/ 单用户openid / 主题Topic
     * @param data 推送的数据
     * @time: 2020/12/16 18:27
     * @author: Rock
     * @version 1.0
     */
    public void sendPushToAccount(int pushType, Object id, JSONObject data) {
        JSONObject tempData = new JSONObject();
        tempData.put("pushType", pushType);
        tempData.put("msgData", data);
        switch (pushType) {
            case ConstKafka.ACCOUNT_TYPE_1:
                tempData.put("groupid", id);
                break;
            case ConstKafka.ACCOUNT_TYPE_2:
                tempData.put("openidList", JSON.toJSONString(id));
                break;
            case ConstKafka.ACCOUNT_TYPE_3:
                tempData.put("openid", id);
                break;
            case ConstKafka.ACCOUNT_TYPE_4:
                tempData.put("topic", id);
                break;
            default:
                log.error("pushType not found. {}", pushType);
                return;
        }
        kafkaProducer.send(id.toString(), tempData.toJSONString());
    }

    /**
     * @param chatGroup
     * @param account
     * @param type
     * @param operatorId
     * @param extOb 拓展的推送字段
     * @time: 2021/1/9 10:33
     * @author: Rock
     * @version 1.0
     */
    public void sendPushToChatGroup(ChatGroup chatGroup, Account account, String type, String operatorId, JSONObject extOb) {
        // 发送kafka
        JSONObject pushData = new JSONObject();
        if(extOb != null){
            pushData = extOb;
        }
        pushData.put("type", type);
        pushData.put("groupid", chatGroup.getId());
        pushData.put("operator", operatorId);
        switch (type) {
            case ConstKafka.CHAT_GROUP_ENTER:
                pushData.put("memberid", StringUtil.isBlank(operatorId) ? account.getId() : operatorId);
                pushData.put("enterTime", DateUtil.nowTimestampToLong());
                pushData.put("phone", StringUtil.isBlank(account.getPhone()) ? "" : account.getPhone());
                pushData.put("name", StringUtil.isBlank(account.getName()) ? "" : account.getName());
                pushData.put("avator", baseBucketTool.getDOMAIN() + BaseBucketTool.ACCOUNT_COMMON_PATH
                        + (StringUtil.isBlank(account.getImage()) ? Const.DEFAULT_JPG : account.getImage()));
                break;
            case ConstKafka.CHAT_GROUP_DISBAND:
                pushData.put("disbandTime", DateUtil.nowTimestampToLong());
                break;
            case ConstKafka.CHAT_GROUP_REMOVE:
                pushData.put("memberid", account.getId());
                pushData.put("removeTime", DateUtil.nowTimestampToDouble());
                break;
            default:
                break;
        }

        sendPushToAccount(ConstKafka.ACCOUNT_TYPE_1, chatGroup.getId(), pushData);
    }

    /**
     * @param imei
     * @param data
     * @time: 2021/1/11 11:47
     * @author: Rock
     * @version 1.0
     */
    public void emitDeviceDemand(String imei, String type, JSONObject data, String openid) {
        if(data == null){
            data = new JSONObject();
        }
        data.put("type", type);
        if(openid != null){
            data.put("openid", openid);
        }
        if (StringUtil.isBlank(data.get("ident"))) {
            data.put("ident", Integer.parseInt(RandomUtil.getRandom(100000, 999999)));
        }
        String onlineStatus = redisClusterService.hget(imei, ConstCacheField.ONLINE_STATUS);
        String kafkaTopicIdent = redisFixService.get(CacheEnum.K_TOPIC_IDENT.getKey() + imei);
        JSONArray tempData = new JSONArray();
        switch (type) {
            case ConstKafka.DEVICE_TYPE_CenterPhone:
                setEmitDeviceDemandCache(imei, "centerPhone", data.toJSONString(), onlineStatus, "centerPhoneSync");
                break;
            case ConstKafka.DEVICE_TYPE_RequestLocation: {
                if(openid.contains("CMS")){
                    //是后台下发的定位,不予标记主动推送
                    break;
                }
                redisClusterService.setex(CacheEnum.RequestLocationIdent.getKey() + imei + ":" + openid, CacheEnum.RequestLocationIdent.getExpire(), openid);
//                data.remove("openid");
            }
            break;
            case ConstKafka.DEVICE_TYPE_DownChatVoice:{
                Device device = deviceService.findById(imei);
                String platform = device.getPlatform();
                String version = device.getVersion();
                if ((StringUtil.isBlank(platform) || "ASR".equals(platform)) && StringUtil.isNumeric(version) && Integer.parseInt(version) < 20201110) {
                    if(data.containsKey("voiceUrl")){
                        String[] tempArr = data.getString("voiceUrl").split("/");
                        String fileName = String.format("http://%s/getway/download/%s", domainName, tempArr[tempArr.length - 1]);
                        data.put("voiceUrl", fileName);
                    }
                }
                //现在无论在线不在线都写到redis, 统一在api处理
                String voiceMsgArr = redisClusterService.hget(imei, ConstCacheField.MESSAGE_VOICE);
                if (StringUtil.isNotBlank(voiceMsgArr)) {
                    try {
                        JSONArray voiceMsgArrJson = JSON.parseArray(voiceMsgArr);
                        if (voiceMsgArrJson != null) {
                            if (voiceMsgArrJson.size() >= 30) {
                                voiceMsgArrJson.remove(0);
                            }
                            tempData.addAll(voiceMsgArrJson);
                        }
                    } catch (Exception e) {
                        log.warn("emitDeviceDemand voiceMsgArr is Exception. imei={}, voiceMsgArr={}", imei, voiceMsgArr);
                        //e.printStackTrace();
                        log.error("旧版微聊语音消息发送出现异常，异常信息为：{}",e.getMessage());
                    }
                }
                tempData.add(data);
                //同步指令缓存
                setEmitDeviceDemandCache(imei, "messageVoice", tempData.toJSONString(), onlineStatus, "messageVoiceSync");
            }
            break;
            case ConstKafka.DEVICE_NEW_APP_VOICE_MSG:{
                //现在无论在线不在线都写到redis, 统一在api处理
                String voiceMsgArr = redisClusterService.hget(imei, ConstCacheField.MESSAGE_VOICE);
                if (StringUtil.isNotBlank(voiceMsgArr)) {
                    try {
                        JSONArray voiceMsgArrJson = JSON.parseArray(voiceMsgArr);
                        if (voiceMsgArrJson != null) {
                            if (voiceMsgArrJson.size() >= 30) {
                                voiceMsgArrJson.remove(0);
                            }
                            tempData.addAll(voiceMsgArrJson);
                        }
                    } catch (Exception e) {
                        log.warn("emitDeviceDemand voiceMsgArr is Exception. imei={}, voiceMsgArr={}", imei, voiceMsgArr);
                        //e.printStackTrace();
                        log.error("新版微聊语音消息发送出现异常，异常信息为：{}",e.getMessage());
                    }
                }
                tempData.add(data);
                //同步指令缓存
                setEmitDeviceDemandCache(imei, "messageVoice", tempData.toJSONString(), onlineStatus, "messageVoiceSync");
            }
            break;
            case ConstKafka.DEVICE_NEW_APP_IMAGE_MSG: {
                //现在无论在线不在线都写到redis, 统一在api处理
                String imageMsgArr = redisClusterService.hget(imei, ConstCacheField.MESSAGE_IMAGE);
                if (StringUtil.isNotBlank(imageMsgArr)) {
                    try {
                        JSONArray imageMsgArrJson = JSON.parseArray(imageMsgArr);
                        if (imageMsgArrJson != null) {
                            if (imageMsgArrJson.size() >= 30) {
                                imageMsgArrJson.remove(0);
                            }
                            tempData.addAll(imageMsgArrJson);
                        }
                    } catch (Exception e) {
                        log.warn("emitDeviceDemand voiceMsgArr is Exception. imei={}, voiceMsgArr={}", imei, imageMsgArr);
                        log.error("新版微聊语音消息发送出现异常，异常信息为：{}",e.getMessage());
                    }
                }
                tempData.add(data);
                setEmitDeviceDemandCache(imei, "messageImage", tempData.toJSONString(), onlineStatus, "messageImageSync");
            }
            break;
            //转发设备上报的语音消息到设备的好友
            case ConstKafka.DEVICE_TYPE_PP_FRIEND_SEND_VOICE:{
                //现在无论在线不在线都写到redis, 统一在api处理
                String ppVoiceMsgArr = redisClusterService.hget(imei, type);
                if (StringUtil.isNotBlank(ppVoiceMsgArr)) {
                    try {
                        JSONArray ppVoiceMsgArrJson = JSON.parseArray(ppVoiceMsgArr);
                        if (ppVoiceMsgArrJson != null) {
                            if (ppVoiceMsgArrJson.size() >= 30) {
                                ppVoiceMsgArrJson.remove(0);
                            }
                            tempData.addAll(ppVoiceMsgArrJson);
                        }
                    } catch (Exception e) {
                        log.warn("emitDeviceDemand ppVoiceMsgArr is Exception. imei={}, ppVoiceMsgArr={}", imei, ppVoiceMsgArr);
                        //e.printStackTrace();
                        log.error("碰碰好友语音消息发送出现异常，异常信息为：{}",e.getMessage());
                    }
                }
                tempData.add(data);
                //同步指令缓存
                setEmitDeviceDemandCache(imei, type, tempData.toJSONString(), onlineStatus, type+"Sync");
            }
            break;
            case ConstKafka.DEVICE_TYPE_Contact: {
                Device device = deviceService.findById(imei);
                if (device == null) {
                    log.error("Device is null. imei={}", imei);
                    break;
                }
//                Integer handleFlag = data.getInteger("handleFlag");
//                data.remove("handleFlag");
                if (Const.INT_1.equals(device.getIsIOTCard())) {
                    //是物联网卡,使用物联网的通讯录
                    IotCardInfo iotCardInfo = iotCardInfoService.findById(device.getIccid());
                    type_16(imei, onlineStatus, iotCardInfo.getContactMap(), data, device.getUserMap());
                } else {
                    type_16(imei, onlineStatus, device.getContactMap(), data, device.getUserMap());
                }
            }
            break;
            //修改碰碰交的好友的图像、电话或名字
            case ConstKafka.DEVICE_TYPE_PP_FRIEND_UPDATE_AVATOR:
            case ConstKafka.DEVICE_TYPE_PP_FRIEND_UPDATE_PHONE_NAME:
            //删除碰碰好友
            case ConstKafka.DEVICE_TYPE_PP_FRIEND_DEL:
            //联系人、新版微聊的微聊联系人
            case ConstKafka.DEVICE_TYPE_ADD_CONTACT:
            case ConstKafka.DEVICE_TYPE_DEL_CONTACT:
            case ConstKafka.DEVICE_TYPE_ADD_NEW_CONTACT:
            case ConstKafka.DEVICE_TYPE_UPDATE_NEW_CONTACT_NAME:
            case ConstKafka.DEVICE_TYPE_UPDATE_NEW_CONTACT_AVATOR:
            case ConstKafka.DEVICE_TYPE_DEL_NEW_CONTACT:
            {
                //旧版增加联系人及删除联系人，新版微聊的增加、修改、删除联系人，单发
                //同步指令
                setEmitDeviceDemandCache(imei, type, data.toJSONString(), onlineStatus, type+"Sync");
                break;
            }
            case ConstKafka.DEVICE_TYPE_SOS:
                setEmitDeviceDemandCache(imei, "sos", data.toJSONString(), onlineStatus, "sosSync");
                break;
//            case ConstKafka.DEVICE_TYPE_:
//                //目标步数
//                setEmitDeviceDemandCache(imei, "targetStep", data.toJSONString(), onlineStatus, "targetStepSync");
//                break;
            case ConstKafka.DEVICE_TYPE_RejectStrangeCallStatus:
                setEmitDeviceDemandCache(imei, "reject", data.toJSONString(), onlineStatus, "rejectSync");
                break;
            case ConstKafka.DEVICE_TYPE_Volume:
                setEmitDeviceDemandCache(imei, "volume", data.toJSONString(), onlineStatus, "volumeSync");
                break;
            case ConstKafka.DEVICE_TYPE_Disturb:
                data.remove("status");
                setEmitDeviceDemandCache(imei, "disBan", data.toJSONString(), onlineStatus, "disBanSync");
                break;
            case ConstKafka.DEVICE_TYPE_Alarm:
                setEmitDeviceDemandCache(imei, "alarms", data.toJSONString(), onlineStatus, "alarmsSync");
                break;
            case ConstKafka.DEVICE_TYPE_ChatMessage:
            case ConstKafka.DEVICE_NEW_APP_TEXT_MSG:
                String textMsgArr = redisClusterService.hget(imei, ConstCacheField.MESSAGE_TEXT);
                if (StringUtil.isNotBlank(textMsgArr)) {
                    try {
                        JSONArray textMsgArrJson = JSON.parseArray(textMsgArr);
                        if (textMsgArrJson != null) {
                            if (textMsgArrJson.size() >= 15) {
                                textMsgArrJson.remove(0);
                            }
                            tempData.addAll(textMsgArrJson);
                        }
                    } catch (Exception e) {
                        log.warn("emitDeviceDemand voiceMsgArr is Exception. imei={}, voiceMsgArr={}", imei, textMsgArr);
                        log.error("微聊文字消息发送出现异常，异常信息为：{}",e.getMessage());
                    }
                }
                tempData.add(data); //最新发的在最后面
                //缓存指令，一旦指令发送失败，等设备再次上线时重新发送缓存的指令
                setEmitDeviceDemandCache(imei, "messageText", tempData.toJSONString(), onlineStatus, "messageTextSync");
                break;
            case ConstKafka.DEVICE_TYPE_LocationUploadInterval:
                setEmitDeviceDemandCache(imei, "locMode_rate", data.toJSONString(), onlineStatus, "locMode_rateSync");
                break;
            case ConstKafka.DEVICE_TYPE_Reset:
                String value = redisClusterService.hget(imei, ConstCacheField.V);
                String deviceLinkPort = redisClusterService.hget(imei, ConstCacheField.DeviceLinkPort);
                String DeviceHead = redisClusterService.hget(imei, ConstCacheField.DeviceHead);
                String shortId = redisClusterService.hget(imei, ConstCacheField.ShortId);
                redisClusterService.del(imei);
                //TODO 记录下发端口的, 清除后需要补上
                if (StringUtil.isNotBlank(deviceLinkPort)) {
                    redisClusterService.hset(imei, ConstCacheField.DeviceLinkPort, deviceLinkPort);
                }
                if (StringUtil.isNotBlank(value)) {
                    redisClusterService.hset(imei, ConstCacheField.V, value);
                } else {
                    redisClusterService.hset(imei, ConstCacheField.V, "20000");
                }
                if (StringUtil.isNotBlank(DeviceHead)) {
                    redisClusterService.hset(imei, ConstCacheField.DeviceHead, DeviceHead);
                }
                if (StringUtil.isNotBlank(shortId)) {
                    redisClusterService.hset(imei, ConstCacheField.ShortId, shortId);
                }
                //最后一次恢复出厂设置的时间
                redisClusterService.hset(imei, ConstCacheField.LAST_RESET_DATA_TIME, String.valueOf(System.currentTimeMillis() / 1000));
                setEmitDeviceDemandCache(imei, "reset", data.toJSONString(), onlineStatus, "resetSync");
                break;
            case ConstKafka.DEVICE_TYPE_AuthCode:
                redisClusterService.hset(imei, ConstCacheField.TOKEN, data.getString("token"));
                break;
            case ConstKafka.DEVICE_TYPE_Inquire:
                redisClusterService.setex(CacheEnum.INQUIRE.getKey() + imei + ":" + data.getString("openid"),
                        CacheEnum.INQUIRE.getExpire(), data.getString("openid"));
                data.remove("openid");
                break;
            case ConstKafka.DEVICE_TYPE_DownPPmessage:
                String offlineMsgArr = redisClusterService.hget(imei, ConstCacheField.PP_OFFLINE_MSG);
                if (StringUtil.isNotBlank(offlineMsgArr)) {
                    try {
                        JSONArray offlineMsgArrJson = JSON.parseArray(offlineMsgArr);
                        if (offlineMsgArrJson != null) {
                            if (offlineMsgArrJson.size() >= 4) {
                                offlineMsgArrJson.remove(0);
                            }
                            tempData.addAll(offlineMsgArrJson);
                        }
                    } catch (Exception e) {
                        log.error("KafkaService Exception={}. imei={}. offlineMsgArr={}", e.getMessage(), imei, offlineMsgArr);
                        try {
                            JSONObject offlineMsgJson = JSON.parseObject(offlineMsgArr);
                            if (offlineMsgJson != null) {
                                tempData.add(offlineMsgJson);
                            }
                        } catch (Exception ee) {
                            log.error("KafkaService Exception={}.", ee.getMessage());
                        }
                    }
                }
                tempData.add(data);
                setEmitDeviceDemandCache(imei, "downPPmessage", tempData.toJSONString(), onlineStatus, "downPPmessageSync");
                break;
//            case ConstKafka.82:
//                //有碰碰交友的设备在修改设备号码时才发送82 指令同步设备电话
//                setEmitDeviceDemandCache(imei, "phone", data.toJSONString(), onlineStatus, "phoneSync");
//                break;
//            case ConstKafka.84:
//                setEmitDeviceDemandCache(imei, "navigation", data.toJSONString(), onlineStatus, "navigationSync");
//                break;
            case ConstKafka.DEVICE_TYPE_DialSwitch:
                setEmitDeviceDemandCache(imei, "dialswitch", data.toJSONString(), onlineStatus, "dialswitchSync");
                break;
//            case ConstKafka.96:
//                redisClusterService.setex(CacheEnum.DATA_FLOWQ.getKey() + imei + ":" + data.getString("openid"),
//                        CacheEnum.DATA_FLOWQ.getExpire(), data.getString("openid"));
//                data.remove("openid");
//                break;
//            case ConstKafka.98:
//                redisClusterService.setex(CacheEnum.DATA_FLOWQ.getKey() + imei + ":" + data.getString("openid"),
//                        CacheEnum.DATA_FLOWQ.getExpire(), data.getString("openid"));
//                data.remove("openid");
//                setEmitDeviceDemandCache(imei, "setDataFlow", data.toJSONString(), onlineStatus, "setDataFlowSync");
//                break;
            case ConstKafka.DEVICE_TYPE_setHeartRate:
            case ConstKafka.DEVICE_TYPE_setBodyTemperature:
                //TODO: 健康功能先不完善
                String redisTempStr = String.format(CacheEnum.UpBodyTemperature.getKey(), imei, data.get("openid"));
                int redisTempExpire = CacheEnum.UpBodyTemperature.getExpire();
                String funDesStr = "体温";
                if(type.equals(ConstKafka.DEVICE_TYPE_setHeartRate)){
                    redisTempStr = String.format(CacheEnum.UpHeartRate.getKey(), imei, data.get("openid"));
                    redisTempExpire = CacheEnum.UpHeartRate.getExpire();
                    funDesStr = "心率";
                }
                if (data.containsKey("frequency")) {
                    Integer frequency = data.getInteger("frequency");
                    if(frequency == 2 || frequency == 1){
                        String tempLogStr = "";
                        if (frequency == 1) {
                            tempLogStr = "开关是关闭的状态, 下发查询一次后,手表上传一次数据后会依旧会关闭心率上传";
                        }
                        log.info("APP 立刻获取一次{}指令下发: {} imei: {} {}",funDesStr, data, imei, tempLogStr);
                        //下发健康功能配置指令(包括立即测量)
                        if (data.containsKey("openid")) {
                            //只有立即测量才有 openid
                            //在redis中保持1分钟,如果有回复,就推送到对应的 openid 的用户APP 上
                            redisClusterService.setex(redisTempStr, redisTempExpire, data.get("openid").toString());
                        }
                    }else if (frequency == 0) {
                        //关闭该功能的
                        log.info("APP 关闭{}上传: {} imei: {}" ,funDesStr, data, imei);
                    }else{
                        //设置频率的
                        log.info("APP 设置{}上传: {} imei: {}" ,funDesStr, data, imei);
                    }
                    setEmitDeviceDemandCache(imei, type, data.toJSONString(), onlineStatus, type+"Sync");
//                    JSONArray tempJsonArr = new JSONArray();
//                    String setHealthArrStr = redisClusterService.hget(imei, ConstCacheField.SET_HEALTH);
//                    if (StringUtil.isNotBlank(setHealthArrStr)) {
//                        tempJsonArr = JSON.parseArray(setHealthArrStr);
//                        int delIndex = -1;
//                        for (int i = 0; i < tempJsonArr.size(); i++) {
//                            if (tempJsonArr.getJSONObject(i) != null
//                                    && tempJsonArr.getJSONObject(i).getInteger("cmdType").equals(data.getInteger("cmdType"))) {
//                                delIndex = i;
//                                break;
//                            }
//                        }
//                        if (delIndex != -1) {
//                            //有重复指令则移除旧的
//                            tempJsonArr.remove(delIndex);
//                        }
//                    }
//                    tempJsonArr.add(data);
//                    setEmitDeviceDemandCache(imei, "setHealth", tempJsonArr.toJSONString(), onlineStatus, "setHealthSync");
                }
                break;
            case ConstKafka.DEVICE_TYPE_setBootOff:
            case ConstKafka.DEVICE_TYPE_opRestart:
            case ConstKafka.DEVICE_TYPE_setSetpSwitch:
            case ConstKafka.DEVICE_TYPE_flower:
            case ConstKafka.DEVICE_TYPE_HEALTH_REMIND:
            case ConstKafka.DEVICE_TYPE_DSAPPFN_DISABLE_OL:
            case ConstKafka.DEVICE_TYPE_SETAPPFN_TIMEOR_DISABLE:
            case ConstKafka.DEV_MESSAGE_SWITCH:
            case ConstKafka.DEV_AUTO_CALL:
            case ConstKafka.PHONE_CHARGE:
            case ConstKafka.DEV_SCREENSAVER:
            case ConstKafka.DEV_TYPE_ACTIVE_FN_INFO:
            case ConstKafka.DEVICE_TYPE_downSchoolTimeTable:
            case ConstKafka.ROLL_TIME_DETECTION:
                //缓存一下指令，等设备上线后重新发送
                setEmitDeviceDemandCache(imei, type, data.toJSONString(), onlineStatus, type+"Sync");
                break;
            //碰碰交好友的指令
            case ConstKafka.DEVICE_TYPE_PP_FRIEND_RESP:
            case ConstKafka.DEVICE_TYPE_PP_FRIEND_STATUS_RESP:
            case ConstKafka.DEV_WIFI_SEARCH:
            case ConstKafka.DEVICE_TYPE_SET_WIFI:
            case ConstKafka.DEVICE_TYPE_DEL_WIFI:
            case ConstKafka.DEVICE_TYPE_CURRENT_CONNECT_WIFI:
            case ConstKafka.POWER_SAVING_STATUS:
            case ConstKafka.SMART_ALARM_STATUS:
            case ConstKafka.SMART_DISTURB_STATUS:
            case ConstKafka.DEVICE_TYPE_DownWeather:
            case ConstKafka.DEVICE_TYPE_REQUEST_CHANGE_ADDR:
            case ConstKafka.DEVICE_TYPE_REQUEST_EXCHANGE_UNBIND:
                break;
//            case 104:
//                setEmitDeviceDemandCache(imei, "monitoringWatchShops", data.toJSONString(), onlineStatus, "monitoringWatchShopsSync");
//                break;
//            case 120:
//                setEmitDeviceDemandCache(imei, "aso", data.toJSONString(), onlineStatus, "asoSync");
//                break;
//            case 122:
//                setEmitDeviceDemandCache(imei, "sedentary", data.toJSONString(), onlineStatus, "sedentarySync");
//                break;
//            case 124:
//                setEmitDeviceDemandCache(imei, "cmsg", data.toJSONString(), onlineStatus, "cmsgSync");
//                break;
//            case 126:
//                setEmitDeviceDemandCache(imei, "powerModelSW", data.toJSONString(), onlineStatus, "powerModelSWSync");
//                break;
//            case 128:
//                setEmitDeviceDemandCache(imei, "lossDev", data.toJSONString(), onlineStatus, "lossDevSync");
//                break;
//            case 130:
//                setEmitDeviceDemandCache(imei, "OTA", data.toJSONString(), onlineStatus, "OTASync");
//                break;
//            case 200:
//                // 在这里 下发通知 设备的 设备QQ 消息虽然不是配置类,但是只需保留最后一条,所以也差不多,作为
//                setEmitDeviceDemandCache(imei, "watchQQLastMsg", data.toJSONString(), onlineStatus, "watchQQLastMsgSync");
//                break;
            default:
                break;
        }
        if ("1".equals(onlineStatus)) {
//            kafkaProducer.sendWatch(imei, data.toJSONString());
            // 为空 或者是 不为数字 发ALL
            if (StringUtil.isBlank(kafkaTopicIdent) || !StringUtil.isNumeric(kafkaTopicIdent)) {
                log.warn("kafkaTopicIdent is null. imei={}", imei);
                kafkaTopicIdent = "ALL";
            }
            kafkaProducer.sendWatch(imei, data.toJSONString(), kafkaTopicIdent);
        } else {
            log.warn("设备未在线，imei: {}", imei);
        }
    }

    /**
     * 类似 设置类 的指令 存储一下在redis里面
     * @param imei
     * @param field
     * @param value
     * @param onlineStatus
     * @param onlineField
     * @time: 2021/2/5 9:58
     * @author: Rock
     * @version 1.0
     */
    private void setEmitDeviceDemandCache(String imei, String field, String value, String onlineStatus, String onlineField) {
        //存放同步指令数据，同步指令所用,设备重新连上线之后（设备发送KI/KA/LA/CONFIG等指令时，会将同步指令一并发下去）
        redisClusterService.hset(imei, field, value);
        //离线的时候设置一个离线标识, 设备会在 登录上线的时候进行同步 (需要在TCP 硬件后台配置文件中设置对应的上线同步指令参数)
        if (!"1".equals(onlineStatus)) {
            redisClusterService.hset(imei, onlineField, "1");
        }
    }

    /**
     * @param imei
     * @param onlineStatus
     * @param valueMap
     * @param data
     * @param memberData 设备绑定的家庭成员
     * @time: 2021/2/5 18:00
     * @author: Rock
     * @version 1.0
     */
    private void type_16(String imei,
                         String onlineStatus,
                         Map<String, Map<String, Object>> valueMap,
                         JSONObject data,
                         Map<String, Map<String, Object>> memberData) {
        // 家庭成员列表
        List<Map<String, Object>> memberList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(memberData)) {
            for (Map.Entry<String, Map<String, Object>> entry : memberData.entrySet()) {
                JSONObject tempJSONOB = new JSONObject(entry.getValue());
                if(!StringUtil.isNumeric(tempJSONOB.getString("phone"))){
                    continue;
                }
                Map<String, Object> temp = new HashMap<>();
                temp.put("id", entry.getKey());
                temp.put("time", tempJSONOB.getInteger("t"));
                temp.put("name", tempJSONOB.getString("relationship"));
                temp.put("phone", tempJSONOB.getString("phone"));
                temp.put("openid", entry.getKey());
                temp.put("relationship_image_id", tempJSONOB.getInteger("relationship_image_id"));
                Account account = accountService.findById(entry.getKey());
                if(account != null){
                    temp.put("url", baseBucketTool.getDOMAIN() + BaseBucketTool.ACCOUNT_COMMON_PATH
                            + (StringUtil.isBlank(account.getImage()) ? Const.DEFAULT_JPG : account.getImage()));
                }
                memberList.add(temp);
            }
            CollectionUtil.sortDouble(memberList, "time", 2);
        }

        List<Map<String, Object>> dataList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(valueMap)) {
            for (Map.Entry<String, Map<String, Object>> entry : valueMap.entrySet()) {
                JSONObject tempJSONOB = new JSONObject(entry.getValue());
                if(tempJSONOB.containsKey("openid")){
                    //如果通讯录里有的 openid 的成员的, 不添加到dataList了, 会和上面的冲突
                    continue;
                }
                if(!StringUtil.isNumeric(tempJSONOB.getString("phone"))){
                    continue;
                }
                Map<String, Object> temp = new HashMap<>();
                temp.put("id", entry.getKey());
                temp.put("time", tempJSONOB.getInteger("t"));
                temp.put("name", tempJSONOB.getString("name"));
                temp.put("phone", tempJSONOB.getString("phone"));
                dataList.add(temp);
            }
            CollectionUtil.sortDouble(dataList, "time", 2);
        }

        List<Map<String, Object>> contactList = new ArrayList<>();
        contactList.addAll(memberList);
        contactList.addAll(dataList);
        data.put("contact", contactList);
        data.put("time", DateUtil.nowTimestampToLong());
        if(data.containsKey("ContactIds")){
            String shortId = RedisClusterService.getJedisCluster().hget(imei, "shortId");
            if (StringUtils.isBlank(shortId)) {
                //非 三基的短标识设备
                data.remove("ContactIds");
            }
        }
        if(data.containsKey("handleFlag")){
            data.remove("handleFlag");
        }
        setEmitDeviceDemandCache(imei, "contacts", data.toJSONString(), onlineStatus, "contactsSync");
    }

    /**
     * @param imei
     * @param onlineStatus
     * @param valueMap
     * @param data
     * @time: 2021/2/6 10:17
     * @author: Rock
     * @version 1.0
     */
    private void contactCacheCheck(String imei, String onlineStatus, Map<String, Map<String, Object>> valueMap, JSONObject data) {
        String contacts = redisClusterService.hget(imei, ConstCacheField.CONTACTS);
        if (StringUtil.isNotBlank(contacts)) {
            JSONObject jsonObject = JSON.parseObject(contacts);
            if (StringUtil.isBlank(jsonObject.get("contact"))) {
                type_16(imei, onlineStatus, valueMap, data, null);
            }
        } else {
            type_16(imei, onlineStatus, valueMap, data, null);
        }
    }

    /**
     * @param imei
     * @param value
     * @time: 2021/5/19 18:05
     * @author: Rock
     * @version 1.0
     */
    public void sendPositionCorrection(String imei, JSONObject value) {
        kafkaProducer.sendPositionCorrection(imei, value.toJSONString());
    }

    public void sendDMPDeviceWithMsg(String imei, JSONObject value) {
        kafkaProducer.sendDMPDeviceWithMsg(imei, value.toJSONString());
    }

}
